package df

import org.apache.spark.SparkConf
import org.apache.spark.sql.{SaveMode, SparkSession}

import java.util.Properties

object E1_DataFrame {
  // 创建 Spark 运行配置对象
  val sparkConf = new SparkConf().setMaster("local[*]").setAppName("E1_DataFrame")
  //创建 SparkSession 对象
  val spark: SparkSession = SparkSession.builder().config(sparkConf).getOrCreate()

  def main(args: Array[String]): Unit = {
    read_jdbc
  }


  def df1() = {
    val data = Seq(("Alice", 30), ("Bob", 25), ("Charlie", 22))
    val columns = Seq("Name", "Age")
    // 导入隐式转换，以便可以使用toDF方法将Seq转换为DataFrame
    import spark.implicits._
    // 将集合转换为DataFrame
    //    spark.createDataFrame(data)
    val df = data.toDF(columns: _*)
    // 显示DataFrame内容
    df.show()
    spark.stop
  }

  def df2_rdd_df = {
    //RDD=>DataFrame=>DataSet 转换需要引入隐式转换规则，否则无法转换
    //spark 不是包名，是上下文环境对象名
    import spark.implicits._

    // 创建RDD（通过集合）
    val rdd = spark.sparkContext.makeRDD(Array(
      (1, "Alice"),
      (2, "Bob"),
      (3, "Charlie"),
      (4, "David"),
      (5, "Eva")))

    // 将RDD转换为DataFrame
    val df = rdd.toDF("id", "name")

    // 显示 DataFrame
    df.show()

    //关闭 Spark
    spark.stop()
  }

  def df3_df_rdd = {
    //RDD=>DataFrame=>DataSet 转换需要引入隐式转换规则，否则无法转换
    //spark 不是包名，是上下文环境对象名
    import spark.implicits._

    // 创建RDD（通过集合）
    val rdd = spark.sparkContext.makeRDD(Array(
      (1, "Alice"),
      (2, "Bob"),
      (3, "Charlie"),
      (4, "David"),
      (5, "Eva")))

    // 将RDD转换为DataFrame
    val df = rdd.toDF("id", "name")

    //***********************************
    // 将DataFrame转换为RDD
    var rdd2 = df.rdd
    //***********************************

    // 显示RDD
    val array: Array[org.apache.spark.sql.Row] = rdd2.collect()
    println(array(0))
    println(array(0)(0))
    println(array(0).getAs[String]("name"))
    //关闭 Spark
    spark.stop()
  }

  def read_json = {
    // 读取 json 文件创建 DataFrame
    var df = spark.read.json(this.getClass.getResource("/sql/user.json").toString)
    df.show()
    // 将 DataFrame 保存到json文件
    df.write.mode("overwrite").json(this.getClass.getResource("/").toString + "output-json")


    // 读取 txt 文件创建 DataFrame
    df = spark.read.text(this.getClass.getResource("/sql/word.txt").toString)
    df.show()
    // 将 DataFrame 保存到 txt 文件
    df.write.mode("overwrite").text(this.getClass.getResource("/").toString + "output-txt")


    // 读取 csv 文件创建 DataFrame
    df = spark.read.option("sep", ",") // 列分隔符
      .option("inferSchema", "true")
      .option("header", "true") // 是否有CSV标头
      //.schema("name STRING, age INT") // 指定列名和类型
      .csv(this.getClass.getResource("/sql/user.csv").toString)
    df.show()
    // 将 DataFrame 保存到 csv 文件
    df.write.mode("overwrite").csv(this.getClass.getResource("/").toString + "output-csv")
  }


  """
    |prepare data
    |
    |mysql
    |create table if not exists user_info(
    |id int,
    |username varchar(20),
    |password varchar(20),
    |age int) engine=InnoDB default charset=utf8;
    |
    |insert into user_info(id, username, password, age) values(1, 'xiaoming', '123', 11),(2, 'xiaoling', '123', 12);
    |""".stripMargin

  def read_jdbc = {
    // 方式1：通用的 load 方法读取
    val df1 = spark.read.format("jdbc")
      .option("url", "jdbc:mysql://localhost:3306/spark")
      .option("driver", "com.mysql.jdbc.Driver")
      .option("user", "root")
      .option("password", "root")
      .option("dbtable", "user_info")
      .load()
    // 显示DataFrame内容
    df1.show()

    // 方式2:通用的 load 方法读取（另一种形式参数）
    val df2 = spark.read.format("jdbc")
      .options(Map(
        "url" -> "jdbc:mysql://localhost:3306/spark?user=root&password=root",
        "dbtable" -> "user_info",
        "driver" -> "com.mysql.jdbc.Driver"))
      .load()
    // 显示DataFrame内容
    df2.show()

    // 方式3:使用 jdbc 方法读取
    val props: Properties = new Properties()
    props.setProperty("user", "root")
    props.setProperty("password", "root")
    val df3 = spark.read.jdbc("jdbc:mysql://localhost:3306/spark", "user_info", props)
    // 显示DataFrame内容
    df3.show()
  }


  def write_jdbc = {
    import spark.implicits._
    // 读取原始数据，创建DataFrame
    val data1 = Seq((3, "小C", "c0c0", 33))
    val data2 = Seq((4, "小D", "d0d0", 3))
    val columns = Seq("id", "username", "password", "age")
    val df1 = data1.toDF(columns: _*)
    val df2 = data2.toDF(columns: _*)

    //方式 1：通用的方式 format 指定写出类型
    df1.write
      .format("jdbc")
      .option("url", "jdbc:mysql://localhost:3306/spark")
      .option("user", "root")
      .option("password", "root")
      .option("dbtable", "user_info")
      .mode(SaveMode.Append)
      .save()

    //方式 2：通过 jdbc 方法
    val props: Properties = new Properties()
    props.setProperty("user", "root")
    props.setProperty("password", "root")
    df2.write.mode(SaveMode.Append).jdbc("jdbc:mysql://localhost:3306/spark", "user_info", props)
  }

  """
    |对 Hive 的支持
    |通过 Hive 创建 DataFrame
    |Spark 支持两种方式来使用 Hive：内置的 Hive 和外部的 Hive。这两种方式在数据处理和元数据存储方面有所不同。
    |
    |内置的 Hive：指的是在 Spark 应用程序中直接使用内置的 Hive Metastore 来存储表的元数据信息，而不依赖外部的 Hive 服务。该方式时候日常的开发学习。
    |外部的 Hive：指的是 Spark 连接到一个独立的外部 Hive Metastore 服务，该服务可能是在独立的 Hive 服务器中或是与 Hadoop 集群中的 Hive Metastore 共享。在这种方式下，Spark 只使用外部的 Hive Metastore 来存储表的元数据信息，数据处理过程与内置的 Hive 没有区别。该方式适用于生产环境。
    |
    |外部，需要启动hive metastore, hive hiveserver2
    |添加配置文件 hive-site.xml, core-site.xml, hdfs-site.xml
    |""".stripMargin
  def read_hive = {
    // 导入隐式转换
    import spark.implicits._

    // 读取原始数据，创建DataFrame
    val data = Seq((1, "老A", 30), (2, "大B", 25), (3, "小C", 22))
    val columns = Seq("id", "name", "age")
    val df = data.toDF(columns: _*)

    // 查看目前所有的Hive表
    spark.sql(
      """
        |create table user_info( id Long, username string, password string, age Long)
        |""".stripMargin)
    spark.sql("show tables").show()
    // 将DataFrame内容写入到Hive表中
    df.write.saveAsTable("user_info")

    // 查看目前所有的Hive表
    spark.sql("show tables").show()

    // 从Hive中读取数据创建DataFrame
    val hiveDF = spark.table("user_info")
    // 显示DataFrame内容
    hiveDF.show()

    //关闭 Spark
    spark.stop()
  }
}
